package com.rraj.common.mq.test;

import javax.annotation.Resource;

import com.rraj.common.mq.MQClient;
import com.rraj.common.mq.MQConstant;
import com.rraj.common.mq.MQProducerMsg;
import com.rraj.common.mq.test.constant.MQTestTagEnum;
import com.rraj.common.mq.test.constant.MQTestTopicEnum;
import com.rraj.common.mq.test.sendCallback.TestSendCallback;
import com.rraj.common.mq.test.transactionExecuter.TestLocalTransactionExecuter;
import com.xiaoleilu.hutool.date.DateUtil;
import org.springframework.stereotype.Service;

/**
 * Created by WangSheng on 2017/6/9.
 *
 * @author WangSheng
 * @date 2017/06/09
 */
@Service
public class MQTestService {

    @Resource
    private MQClient mqClient;

    @Resource
    private TestLocalTransactionExecuter testLocalTransactionExecuter;
    @Resource
    private TestSendCallback testSendCallback;

    /**
     * 同步发送普通消息
     */
    public boolean sendCommonMessageSync() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_DEFAULT.getTopic());
            msg.setTag(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_SYNC.getValue());
            msg.setKey(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_SYNC.getValue() + i);
            msg.setMsgBody(("同步发送普通消息：" + i).getBytes());
            mqClient.sendCommonMessageSync(msg);
        }
        return true;
    }

    /**
     * 异步发送普通消息
     */
    public boolean sendCommonMessageAsync() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_DEFAULT.getTopic());
            msg.setTag(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_ASYNC.getValue());
            msg.setKey(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_ASYNC.getValue() + i);
            msg.setMsgBody(("异步发送普通消息：" + i).getBytes());
            mqClient.sendCommonMessageAsync(msg, testSendCallback);
        }
        return true;
    }

    /**
     * 单向发送普通消息
     */
    public boolean sendCommonMessageOneway() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_DEFAULT.getTopic());
            msg.setTag(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_ONEWAY.getValue());
            msg.setKey(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_ONEWAY.getValue() + i);
            msg.setMsgBody(("单向发送普通消息：" + i).getBytes());
            mqClient.sendCommonMessageOneway(msg);
        }
        return true;
    }

    /**
     * 发送定时消息
     */
    public boolean sendTimingMessage() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_DEFAULT.getTopic());
            msg.setTag(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_TIMING.getValue());
            msg.setKey(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_TIMING.getValue() + i);
            msg.setMsgBody(("发送定时消息：" + i).getBytes());
            mqClient.sendTimingMessage(msg, DateUtil.parse("2017-06-12 11:36:00"));
        }
        return true;
    }

    /**
     * 发送延时消息
     */
    public boolean sendDelayMessage() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_DEFAULT.getTopic());
            msg.setTag(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_DELAY.getValue());
            msg.setKey(MQTestTagEnum.TAG_DEFAULT_TEST_TOPIC1_DELAY.getValue() + i);
            msg.setMsgBody(("发送延时消息：" + i).getBytes());
            mqClient.sendDelayMessage(msg, 30000);
        }
        return true;
    }

    /**
     * 发送事务消息
     */
    public boolean sendTransactionMessage() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_TRANSACTION.getTopic());
            msg.setTag(MQTestTagEnum.TAG_TRANSACTION_TEST_TOPIC1_TAG1.getValue());
            msg.setKey(MQTestTagEnum.TAG_TRANSACTION_TEST_TOPIC1_TAG1.getValue() + i);
            msg.setMsgBody(("发送事务消息：" + i).getBytes());
            mqClient.sendTransactionMessage(msg, testLocalTransactionExecuter,"testTransactionChecker", null);
        }
        return true;
    }

    /**
     * 发送全局顺序消息
     */
    public boolean sendOrderAllMessage() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_ALL_ORDER.getTopic());
            msg.setTag(MQTestTagEnum.TAG_ALL_ORDER_TEST_TOPIC1_TAG1.getValue());
            msg.setKey(MQTestTagEnum.TAG_ALL_ORDER_TEST_TOPIC1_TAG1.getValue() + i);
            msg.setMsgBody(("发送全局顺序消息：" + i).getBytes());
            mqClient.sendOrderAllMessage(msg);
        }
        return true;
    }

    /**
     * 发送分区顺序消息
     */
    public boolean sendOrderShardingMessage() throws Exception {
        for (int i = 0; i < 10; i++) {
            MQProducerMsg msg = new MQProducerMsg();
            msg.setTopic(MQTestTopicEnum.TOPIC_SHARDING_ORDER.getTopic());
            msg.setTag(MQTestTagEnum.TAG_SHARDING_ORDER_TEST_TOPIC1_TAG1.getValue());
            msg.setKey(MQTestTagEnum.TAG_SHARDING_ORDER_TEST_TOPIC1_TAG1.getValue() + i);
            msg.setMsgBody(("发送分区顺序消息：" + i).getBytes());
            mqClient.sendOrderShardingMessage(msg, MQConstant.SHARDING_ALL_ORDER_TEST_TOPIC1_SHARDING1);
        }
        return true;
    }
}
